package com.bfxy.rabbit.producer.broker;

import com.bfxy.rabbit.api.enums.MessageTypeEnum;
import com.bfxy.rabbit.api.model.Message;
import com.bfxy.rabbit.producer.constant.BrokerMessageConst;
import com.bfxy.rabbit.producer.enums.BrokerMessageStatusEnum;
import com.bfxy.rabbit.producer.entity.BrokerMessage;
import com.bfxy.rabbit.producer.service.MessageStoreService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;

/**
 * $RabbitBrokerImpl 真正的发送不同类型的消息实现类
 */
@Component
@Slf4j
public class RabbitBrokerImpl implements RabbitBroker {

//    @Resource
//    private RabbitTemplate rabbitTemplate;

    @Resource
    private RabbitTemplateContainer rabbitTemplateContainer;

    @Resource
    private MessageStoreService messageStoreService;

    /**
     * 迅速发消息
     * @param message
     */
    @Override
    public void rapidSend(final Message message) {
        message.setMessageType(MessageTypeEnum.RAPID);
        sendKernel(message);
    }

    @Override
    public void confirmSend(final Message message) {
        message.setMessageType(MessageTypeEnum.CONFIRM); // RabbitTemplateContainer里对消息的confirm确认方法进行了封装
        sendKernel(message);
    }

    @Override
    public void reliantSend(final Message message) {
        message.setMessageType(MessageTypeEnum.RELIANT);

        // 判断数据库中该条消息记录是否存在
        BrokerMessage brokerMessageDB = messageStoreService.selectByMessageId(message.getMessageId());
        if (brokerMessageDB == null) {
            // 1.记录数据库消息发送的日志
            Date now = new Date();
            BrokerMessage brokerMessage = new BrokerMessage();
            brokerMessage.setMessageId(message.getMessageId());
            brokerMessage.setStatus(BrokerMessageStatusEnum.SENDING.getCode());
            // tryCount默认是0，在最开始发送的时候不需要设置
            //brokerMessage.setTryCount();
            brokerMessage.setNextRetry(DateUtils.addMinutes(now, BrokerMessageConst.TIMEOUT));
            brokerMessage.setCreateTime(now);
            brokerMessage.setUpdateTime(now);
            brokerMessage.setMessage(message);
            messageStoreService.insert(brokerMessage);
        }

        // 2.执行真正的发送消息逻辑
        sendKernel(message);
    }

    @Override
    public void sendMessages() {
        List<Message> messageList = MessageHolder.clear();
        messageList.forEach(message -> {
            MessageHolderAsyncQueue.submit(new Runnable() {
                @Override
                public void run() {
                    CorrelationData correlationData = new CorrelationData(String.format("%s#%s#%s", message.getMessageId(), System.currentTimeMillis(), message.getMessageType().getValue()));
                    String topic = message.getTopic();
                    String routingKey = message.getRoutingKey();
                    RabbitTemplate rabbitTemplate = rabbitTemplateContainer.getRabbitTemplate(message);
                    rabbitTemplate.convertAndSend(topic, routingKey, message, correlationData);
                    log.info("#RabbitBrokerImpl.sendMessages# send to rabbitmq, messageId: {}", message.getMessageId());
                }
            });
        });
    }

    /**
     * 发送消息的核心方法，使用异步线程池发送消息
     * @param message
     */
    private void sendKernel(Message message) {
        AsyncBaseQueue.submit(new Runnable() {
            @Override
            public void run() {
                CorrelationData correlationData = new CorrelationData(String.format("%s#%s#%s", message.getMessageId(), System.currentTimeMillis(), message.getMessageType().getValue()));
                String topic = message.getTopic();
                String routingKey = message.getRoutingKey();
                RabbitTemplate rabbitTemplate = rabbitTemplateContainer.getRabbitTemplate(message);
                rabbitTemplate.convertAndSend(topic, routingKey, message, correlationData);
                log.info("#RabbitBrokerImpl.sendKernel# send to rabbitmq, messageId: {}", message.getMessageId());
            }
        });
    }

}
